package cn.panshi.cdcdemo;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import org.bson.Document;
import org.springframework.stereotype.Component;

import com.mongodb.client.model.changestream.OperationType;

import cn.panshi.cdcdemo.MongoCdcBizRouter.MongoCdcEnum;
import cn.panshi.mongocdc.BizRouterAnnotation;
import cn.panshi.mongocdc.IMongoCdcBizRouter;
import cn.panshi.mongocdc.ReflectionUtils;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class MongoCdcBizRouter implements IMongoCdcBizRouter<MongoCdcEnum> {


	@BizRouterAnnotation(enumClazz = MongoCdcEnum.class, enumValue = "BizFlowDataCollectionPlan")
	private void doNotifyFollowers4NewFlowData(String bizId, String resumeToken,
			Document flowCollectionDoc) {
		log.debug("BizFlowDataCollectionPlanDoc cdc " + flowCollectionDoc);


	}

	@BizRouterAnnotation(enumClazz = MongoCdcEnum.class, enumValue = "Test")
	private void test(String bizId, String resumeToken, Document doc) {
		if (doc != null) {
			log.debug("BizFlowDataCollectionPlanDoc cdc " + doc);
		} else {
			log.debug("_id:" + bizId);
		}
	}

	public enum MongoCdcEnum {
		Coll1("Coll1",
				new OperationType[] { OperationType.INSERT, OperationType.UPDATE }), //
		Coll2("Coll2", new OperationType[] { OperationType.INSERT }), //

		;

		private String collectionName;

		private OperationType[] operationType;

		private MongoCdcEnum(String collectionName, OperationType[] operationType) {
			this.collectionName = collectionName;
			this.operationType = operationType;
		}

		public String getCollectionName() {
			return collectionName;
		}

		public OperationType[] getOperationType() {
			return operationType;
		}

		public static MongoCdcEnum match(String collectionName, OperationType operationType) {
			List<MongoCdcEnum> enumValues = Arrays.asList(MongoCdcEnum.values());
			List<MongoCdcEnum> matched = enumValues.stream()//
					.filter(each -> each.getCollectionName().equals(collectionName)
							&& Arrays.asList(each.getOperationType()).contains(operationType))//
					.collect(Collectors.toList());

			return matched.isEmpty() ? null : matched.get(0);
		}

	}

	@Override
	public Class<MongoCdcEnum> getBizEnumType() {
		return MongoCdcEnum.class;
	}

	@Override
	public MongoCdcEnum[] getBizEnumValues() {
		return MongoCdcEnum.values();
	}

	@Override
	public MongoCdcEnum parse(String collectionName, OperationType operationType) {
		return MongoCdcEnum.match(collectionName, operationType);
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	@Override
	public Method matchBizMethodByCdc(Enum cdc) {
		List<Method> methodsWithAnnotation = ReflectionUtils.getMethodWithAnnotation(this.getClass(),
				BizRouterAnnotation.class);

		List<Object> matchedMethods = new ArrayList<>();
		methodsWithAnnotation.stream().filter(m -> {
			Class<? extends Enum> annType = m.getAnnotation(BizRouterAnnotation.class).enumClazz();
			String annValue = m.getAnnotation(BizRouterAnnotation.class).enumValue();
			Enum annTypeEnum = Enum.valueOf(annType, annValue);
			return annTypeEnum == cdc;
		}).forEach(each -> matchedMethods.add(each));

		if (matchedMethods.isEmpty()) {
			log.error("No method found with MongoCdcBizAnnotation(" + cdc.name() + ")");
			return null;
		} else {
			return (Method) matchedMethods.get(0);
		}

	}
}